package com.nulldev.util.networking.tcp_legacy;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.regex.Pattern;

import com.nulldev.util.JVM.JVM;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutorManager;
import com.nulldev.util.io.IOUtils;
import com.nulldev.util.logging.Logger;
import com.nulldev.util.logging.LoggerFactory;

abstract public class TCPCore {
	static private final byte[] empty = new byte[0];
	public final Logger loggerF = LoggerFactory.getLogger();
	final String category;
	private final String name;
	private final Socket socket;
	// final LargeDataInputStream input;
	final DataInputStream input;
	// final LargeDataOutputStream output; //DataOutputStream output;
	final DataOutputStream output;
	final Object outputLock = new Object();

	final ArrayBlockingQueue<String> sends = new ArrayBlockingQueue<String>(1024, true);
	volatile Thread writeThread;
	public volatile boolean closed;
	volatile byte[] data = empty;

	public TCPCore(String category, String name, final Socket socket) throws IOException {
		this.category = category;
		this.name = name;
		this.socket = socket;
		// this.loggerF.traceEnabled = true;

		try {
			input = new DataInputStream(socket.getInputStream());
			output = new DataOutputStream(socket.getOutputStream());
			// input = new LargeDataInputStream(socket.getInputStream());
			// output = new LargeDataOutputStream(socket.getOutputStream()); //new
			// DataOutputStream(socket.getOutputStream());
		} catch (IOException ex) {
			throw new IOException("Error opening socket streams.", ex);
		}
	}

	private volatile boolean wantsToDisconnect;

	public void requestDisconnect() {
		this.wantsToDisconnect = true;
		int i = 3600;
		while (i > 0) {
			try {
				final boolean closed = close();
				if (closed) {
					break;
				}
				Thread.sleep(50L);
				i -= 50;
			} catch (Exception e) {
			}
		}
	}

	public Socket getSocket() {
		return this.socket;
	}

	public void start() {
		final Thread readThread = new Thread(name + "Read") {
			@SuppressWarnings("deprecation")
			public void run() {
				try {
					while (!closed) {
						String message = "";
						try {
							message = input.readUTF();
						} catch (Exception e) {
							try {
								if (!(e instanceof EOFException) && (!(e instanceof SocketException) && !e.getMessage().equals("Socket closed"))) {
									loggerF.error("read() -> Failed to get UTF8 string, falling back...", e);
								}
								message = input.readLine();
							} catch (Exception ea) {
								loggerF.error(ea);
							}
						}
						if (message == null || closed)
							break;

						long readTime = System.currentTimeMillis();

						String event, payload;
						int index = message.indexOf(" ");
						if (index != -1) {
							event = message.substring(0, index).trim();
							payload = message.substring(index + 1).trim();
						} else {
							event = message.trim();
							payload = "";
						}

						// int dataLength = TCPUtil.readInt(input);
						final long dataLength = TCPUtil.readLong(input);
						if (dataLength > 0) {
							if (closed)
								break;
							// TODO: Allow LONG reading.
							if (data.length < dataLength)
								data = new byte[(int) dataLength];
							int offset = 0, remaining = (int) dataLength;
							while (true) {
								int count = input.read(data, offset, remaining);
								if (count == -1 || closed)
									break;
								remaining -= count;
								if (remaining == 0)
									break;
								offset += count;
							}
						}

						readTime = System.currentTimeMillis() - readTime;

						if (loggerF.traceEnabled.get())
							loggerF.trace(category, "Received: " + event + ", " + payload + (dataLength > 0 ? ", " + dataLength : ""));
						try {
							receive(event.replaceAll(Pattern.quote("%space"), " "), payload, data, dataLength, readTime);
						} catch (Throwable ex) {
							if (loggerF.errorEnabled.get())
								loggerF.error(category, "Error processing message: " + (message.length() < 320 ? message : "size=" + message.length()), ex);
							break;
						}
					}
				} catch (EOFException ex) {
					if (loggerF.traceEnabled.get())
						loggerF.trace(category, "Connection has closed.", ex);
				} catch (IOException ex) {
					if (!closed) {
						if (ex.getMessage() != null && ex.getMessage().contains("Connection reset")) {
							if (loggerF.traceEnabled.get())
								loggerF.trace(category, "Client connection reset.", ex);
						} else {
							if (loggerF.errorEnabled.get())
								loggerF.error(category, "Error reading from connection.", ex);
						}
					}
				} catch (Exception ex) {
					if (loggerF.errorEnabled.get())
						loggerF.error(category, "Error occured.", ex);
				} finally {
					close();
					if (loggerF.traceEnabled.get())
						loggerF.trace(category, "Read thread stopped.");
				}
			}
		};
		if (this.PREFER_THREAD_API) {
			ExecutorManager.get().queueTask(readThread);
		} else {
			readThread.start();
		}

		this.writeThread = new Thread(name + "Write") {
			public void run() {
				try {
					while (!closed) {
						try {
							String take = sends.take();
							if (!take.endsWith(JVM.lineSeperator())) {
								take = take + JVM.lineSeperator();
							}
							sendBlocking(take, null, 0, 0);
						} catch (InterruptedException ignored) {
						}
					}
				} finally {
					close();
					if (loggerF.traceEnabled.get())
						loggerF.trace(category, "Write thread stopped.");
				}
			}
		};
		if (this.PREFER_THREAD_API) {
			ExecutorManager.get().queueTask(this.writeThread);
		} else {
			this.writeThread.start();
		}
	}

	public boolean PREFER_THREAD_API;

	/** Sends the string without waiting for the send to complete. */
	public void send(String message) {
		if (loggerF.traceEnabled.get())
			loggerF.trace(category, "Queued: " + message);
		sends.add(message);
	}

	/**
	 * Sends the string, blocking until sending is complete.
	 * 
	 * @return false if the connection is closed or the send failed (which closes
	 *         the connection).
	 */
	public boolean sendBlocking(String message) {
		return sendBlocking(message, null, 0, 0);
	}

	/** @see #sendBlocking(String, byte[], int, int) */
	public boolean sendBlocking(String message, byte[] bytes) {
		return sendBlocking(message, bytes, 0, bytes.length);
	}

	/**
	 * Sends the string, blocking until sending is complete.
	 * 
	 * @return false if the connection is closed or the send failed (which closes
	 *         the connection).
	 */
	public boolean sendBlocking(String message, byte[] bytes, int offset, int count) {
		if (message == null) {
			loggerF.error("Failed to send message. [messageNull=" + (message == null) + ",bytesNull=" + (bytes == null) + "]");
			return false;
		}
		message = message.replaceAll(Pattern.quote(" "), "%space");
		if (closed)
			return false;
		try {
			synchronized (outputLock) {
				if (loggerF.traceEnabled.get())
					loggerF.trace(category, "Sent: " + message + (count > 0 ? ", " + count : ""));
				if (closed)
					return false;
				try {
					output.writeUTF(message);
					output.flush();
					// TCPUtil.writeInt(count, output);
					TCPUtil.writeLong(count, output);
					output.flush();
					output.write(bytes, offset, count);
					output.flush();

					if (wantsToDisconnect) {
						close();
						wantsToDisconnect = false;
					}
				} catch (Exception e) {
					loggerF.error("Peer disconnected before being able to write.", e);
				}
			}
			return true;
		} catch (Exception ex) {
			if (loggerF.errorEnabled.get() && !closed)
				loggerF.error(category, "Error writing to connection.", ex);
			close();
			return false;
		}
	}

	abstract public void receive(final String event, final String payload, final byte[] bytes, final long count, final long readTime);

	public boolean close() {
		if (loggerF.traceEnabled.get() && !closed)
			loggerF.trace(category, "Client disconnected.");
		closed = true;
		if (writeThread != null)
			writeThread.interrupt();
		IOUtils.closeQuietly(output);
		IOUtils.closeQuietly(input);
		IOUtils.closeQuietly(socket);
		return true;
	}

	public Logger getLogger() {
		return this.loggerF;
	}

	public String format() {
		return this.socket.getInetAddress().getHostAddress() + ":" + this.socket.getPort();
	}
}
